package cn.lagou.homework

import org.apache.spark.sql.SparkSession
import org.apache.spark.SparkConf

object HomeWork1DSL_ip {
  def main(args: Array[String]): Unit = {
    // 初始化
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName).setMaster("local[*]")
    val spark: SparkSession = SparkSession.builder()
      .config(conf)
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")

    // 读数据，并解析。将ip地址转换为Long类型
    import spark.implicits._
    spark.read
      .option("delimiter", "|")
      .csv("data/http.log")
      .map(row => row.getString(1))
      .createOrReplaceTempView("t1")

    // 读数据，解析，收回来。最后变为广播变量
    val ipData: Array[(Long, Long, String)] = spark.read
      .option("delimiter", "|")
      .csv("data/ip.dat")
      .map(row => (row.getString(2).toLong, row.getString(3).toLong, row.getString(6)))
      .collect()
    val ipBC = spark.sparkContext.broadcast(ipData.sortBy(_._1))

    def ip2Long(ip: String): Long = {
      ip.split("\\.")
        .map(_.toLong)
        .fold(0L) { (buffer, elem) =>
          buffer << 8 | elem
        }
    }

    // 给定ip地址，在ips中查找对应的城市名。使用二分查找算法
    def getCityName(ip: Long): String = {
      val ips: Array[(Long, Long, String)] = ipBC.value
      var start = 0
      var end = ips.length - 1
      var middle = 0

      while (start <= end) {
        middle = (start + end) / 2
        if ((ip >= ips(middle)._1) && (ip <= ips(middle)._2))
          return ips(middle)._3
        else if (ip < ips(middle)._1)
          end = middle - 1
        else
          start = middle + 1
      }
      "Unknown"
    }

    spark.udf.register("ip2Long", ip2Long _)
    spark.udf.register("getCityName", getCityName _)
    spark.sql(
      """
        |select getCityName(ip2Long(value)) as provice, count(1) as no
        |  from t1
        |group by getCityName(ip2Long(value))
        |""".stripMargin).show

    spark.close()
  }
}